package com.arch.bigdata.hbase1x.nx;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * 作者： 马中华   https://blog.csdn.net/zhongqi2513
 * 时间： 2017/11/16 16:45
 * 描述： 测试HBase的分页查询
 */
public class Hbase_Page01 {

    private static final String ZK_CONNECT_STR = "bigdata02:2181,bigdata03:2181,bigdata04:2181";

    private static final String TABLE_NAME = "user_info";
    private static final String FAMILY_BASIC = "base_info";
    private static final String FAMILY_EXTRA = "extra_info";
    private static final String COLUMN_NAME = "name";
    private static final String COLUMN_AGE = "age";
    private static final String ROW_KEY = "rk0001";

    private static Configuration config = null;
    private static HTable table = null;

    static {
        config = HBaseConfiguration.create();
        config.set("hbase.zookeeper.quorum", ZK_CONNECT_STR);
        try {
            table = new HTable(config, TABLE_NAME);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public static void main(String[] args) throws Exception {
        ResultScanner pageData = getPageData(2, 3);
        HBasePrintUtil.printResultScanner(pageData);
    }

    public static ResultScanner getPageData(int pageIndex, int pageNumber) throws Exception {
        // 怎么把pageIndex 转换成 startRow
        String startRow = null;

        if (pageIndex == 1) { // 当客户方法只取第一页的分页数据时，
            ResultScanner pageData = getPageData(startRow, pageNumber);
            return pageData;

        } else {
            ResultScanner newPageData = null;
            for (int i = 0; i < pageIndex - 1; i++) { // 总共循环次数是比你取的页数少1
                newPageData = getPageData(startRow, pageNumber);
                startRow = getLastRowkey(newPageData);
                byte[] add = Bytes.add(Bytes.toBytes(startRow), new byte[]{0X00});
                startRow = Bytes.toString(add);
            }
            newPageData = getPageData(startRow, pageNumber);
            return newPageData;
        }
    }

    /**
     * scan 'user_info',{COLUMNS => 'base_info:name',LIMIT => 4, STARTROW => 'zhangsan_20150701_0001'}
     */
    public static ResultScanner getPageData(String startRow, int pageNumber) throws Exception {
        Scan scan = new Scan();
        scan.addColumn(Bytes.toBytes("base_info"), Bytes.toBytes("name"));
        // 設置當前查询的其实位置
        if (!StringUtils.isBlank(startRow)) {
            scan.setStartRow(Bytes.toBytes(startRow));
        }
        // 第二个参数
        Filter pageFilter = new PageFilter(pageNumber);
        scan.setFilter(pageFilter);

        ResultScanner rs = table.getScanner(scan);
        return rs;
    }

    public static String getLastRowkey(ResultScanner rs) {
        String lastRowkey = null;
        for (Result result : rs) {
            lastRowkey = Bytes.toString(result.getRow());
        }
        return lastRowkey;
    }
}